package cn.sheep.dolphin.tag

import cn.sheep.dolphin.common.{DolphinAppComm, Jedispools}
import cn.sheep.dolphin.utils.TagsUtils
import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext

/** 用户的最终标签数据聚合 - graph
  * author: old sheep
  * QQ: 64341393 
  * Created 2018/12/5
  */
object UserTagsGraph {

	def main(args: Array[String]): Unit = {
		if (args.size != 3) {
			println(
				"""
				  |Usage: cn.sheep.dolphin.tag.UserTagsGraph
				  |	<parquetInputPath, tagOutputPath, stopwordPath>
				""".stripMargin)
			sys.exit(-1)
		}

		// 接受参数
		val Array(parquetInputPath, tagOutputPath, stopwordPath) = args

		val sc = DolphinAppComm.createSparkContext("用户的最终标签数据聚合")
		// 广播停用词字典数据
		val stopWords = sc.textFile(stopwordPath).map((_, null)).collectAsMap()
		val stopwordBT = sc.broadcast(stopWords)


		val sQLContext = new SQLContext(sc)
		val dataFrame = sQLContext.read.parquet(parquetInputPath)

		val filtered = dataFrame.filter(
			"""
			  |imei!='' or idfa!='' or mac!='' or androidid!='' or openudid!='' or
			  |imeimd5!='' or idfamd5!='' or macmd5!='' or androididmd5!='' or openudidmd5!='' or
			  |imeisha1!='' or idfasha1!='' or macsha1!='' or androididsha1!='' or openudidsha1!=''
			""".stripMargin)

		// 整理RDD[(UserIds, Tags)]
		val markedData = filtered.mapPartitions(partition => {
			// jedis
			val jedis = Jedispools.getJedis()
			val resultSet = partition.map(row => {
				val tagsMap = Tags4Device.makeTags(row, jedis) ++ Tags4Ads.makeTags(row, stopwordBT)
				(TagsUtils.getAllUserId(row), tagsMap)
			})
			jedis.close()
			resultSet
		})

		// 点集合RDD[(Long, VD=Tags+UserId)]
		val userVertices: RDD[(Long, (String, Map[String, Int]))] = markedData.flatMap(tp => {
			val headId = tp._1.head
			tp._1.map(uId => {
				if (uId.equals(headId)) (uId.hashCode.toLong, (uId, tp._2))
				else (uId.hashCode.toLong, (uId, Map.empty[String, Int]))
			})
		})

		// 边集合
		val userEdges: RDD[Edge[Null]] = markedData.flatMap(tp => {
			val headId = tp._1.head.hashCode.toLong
			tp._1.map(uId => Edge(headId, uId.hashCode.toLong, null))
		})

		// 图对象
		val graph = Graph(userVertices, userEdges)

		// 找关系(a, commId), (b, commId)
		val cc = graph.connectedComponents().vertices

		cc.join(userVertices).map {
			case (_, (minId, (uId, tagsMap))) => (minId, (Set(uId), tagsMap.toList))
		}.reduceByKey((a, b) => {
			val userIds = a._1 ++ b._1
			val tags = (a._2 ++ b._2).groupBy(_._1).mapValues(_.map(_._2).sum).toList
			(userIds, tags)
		}).map {
			case (_, (userIds, tagsIter)) => userIds.mkString(",") + "\t" + tagsIter.mkString("$")
		}.saveAsTextFile(tagOutputPath)

		sc.stop()

	}

}
